package com.ehualu.impda.hadoop;

import com.ehualu.impda.kafkasecurity.LoginUtilKafkaHuawei;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ExecutionException;


public class NewProducer2 extends Thread {
    private static final Logger LOG = LoggerFactory.getLogger(NewProducer2.class);


    private static final String USER_KEYTAB_FILE = "user.keytab";

    /**
     * 用户自己申请的机机账号名称
     */
    private static final String USER_PRINCIPAL = "super_rj";

    /**
     * 新Producer 构造函数
     *
     * @param topicName Topic名称
     */

    /**
     * 生产者线程执行函数，循环发送消息。
     */

    public static void securityPrepare() throws IOException {
        String filePath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "main" + File.separator + "resources" + File.separator;
        String krbFile = filePath + "krb5.conf";
        String userKeyTableFile = filePath + USER_KEYTAB_FILE;

        //windows路径下分隔符替换
        userKeyTableFile = userKeyTableFile.replace("\\", "\\\\");
        krbFile = krbFile.replace("\\", "\\\\");

        LoginUtilKafkaHuawei.setKrb5Config(krbFile);
        LoginUtilKafkaHuawei.setZookeeperServerPrincipal("zookeeper/hadoop.hadoop.com");
        LoginUtilKafkaHuawei.setJaasFile(USER_PRINCIPAL, userKeyTableFile);
    }

    public static Boolean isSecurityModel() {
        Boolean isSecurity = false;
        String krbFilePath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "main" + File.separator + "resources" + File.separator + "kafkaSecurityMode";

        Properties securityProps = new Properties();

        // file does not exist.
        if (!isFileExists(krbFilePath)) {
            return isSecurity;
        }

        try {
            securityProps.load(new FileInputStream(krbFilePath));
            if ("yes".equalsIgnoreCase(securityProps.getProperty("kafka.client.security.mode"))) {
                isSecurity = true;
            }
        } catch (Exception e) {
            LOG.info("The Exception occured : {}.", e);
        }

        return isSecurity;
    }

    /*
     * 判断文件是否存在
     */
    private static boolean isFileExists(String fileName) {
        File file = new File(fileName);

        return file.exists();
    }

    public static void main(String[] args) {


        String bootstrapServers = "bootstrap.servers";


        String clientId = "client.id";


        String keySerializer = "key.serializer";


        String valueSerializer = "value.serializer";

        String securityProtocol = "security.protocol";


        String saslKerberosServiceName = "sasl.kerberos.service.name";


        String kerberosDomainName = "kerberos.domain.name";


        Properties props = new Properties();

        KafkaProperties kafkaProc = KafkaProperties.getInstance();

        // Broker地址列表
        props.put(bootstrapServers, kafkaProc.getValues(bootstrapServers, "37.158.97.137:21005,37.158.97.136:21005"));
        // 客户端ID
        props.put(clientId, kafkaProc.getValues(clientId, "DemoProducer"));
        // Key序列化类
        props.put(keySerializer,
                kafkaProc.getValues(keySerializer, "org.apache.kafka.common.serialization.IntegerSerializer"));
        // Value序列化类
        props.put(valueSerializer,
                kafkaProc.getValues(valueSerializer, "org.apache.kafka.common.serialization.StringSerializer"));
        // 协议类型:当前支持配置为SASL_PLAINTEXT或者PLAINTEXT
        props.put(securityProtocol, kafkaProc.getValues(securityProtocol, "SASL_PLAINTEXT"));
        // 服务名
        props.put(saslKerberosServiceName, "kafka");
        // 域名
        props.put(kerberosDomainName, kafkaProc.getValues(kerberosDomainName, "hadoop.hadoop.com"));

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        String topic = "gaoshujson";


        if (isSecurityModel()) {
            try {
                LOG.info("Securitymode start.");

                //!!注意，安全认证时，需要用户手动修改为自己申请的机机账号
                securityPrepare();


                int messageNo = 1;
                // 指定发送多少条消息后sleep1秒
                int intervalMessages = 10;

                while (messageNo <= 100) {
                    String messageStr = "Message_" + messageNo;
                    long startTime = System.currentTimeMillis();

                    // 构造消息记录
                    ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, messageStr);


                    try {
                        // 同步发送
                        producer.send(record).get();
                    } catch (InterruptedException ie) {
                        LOG.info("The InterruptedException occured : {}.", ie);
                    } catch (ExecutionException ee) {
                        LOG.info("The ExecutionException occured : {}.", ee);
                    }


                    messageNo++;

                    if (messageNo % intervalMessages == 0) {
                        // 每发送intervalMessage条消息sleep1秒
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        LOG.info("The Producer have send {} messages.", messageNo);
                    }
                }


            } catch (IOException e) {
                LOG.error("Security prepare failure.");
                LOG.error("The IOException occured.", e);
                return;
            }

            System.out.println("send message over.");
            producer.close();

            LOG.info("Security prepare success.");
        }


    }
}
//
//class DemoCallBack implements Callback {
//    private static Logger LOG = LoggerFactory.getLogger(DemoCallBack.class);
//
//    private long startTime;
//
//    private int key;
//
//    private String message;
//
//    public DemoCallBack(long startTime, int key, String message) {
//        this.startTime = startTime;
//        this.key = key;
//        this.message = message;
//    }
//
//    /**
//     * 回调函数，用于处理异步发送模式下，消息发送到服务端后的处理。
//     *
//     * @param metadata  元数据信息
//     * @param exception 发送异常。如果没有错误发生则为Null。
//     */
//    @Override
//    public void onCompletion(RecordMetadata metadata, Exception exception) {
//        long elapsedTime = System.currentTimeMillis() - startTime;
//        if (metadata != null) {
//            LOG.info("message(" + key + ", " + message + ") sent to partition(" + metadata.partition() + "), "
//                    + "offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
//        } else if (exception != null) {
//            LOG.error("The Exception occured.", exception);
//        }
//
//    }
//}